import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import BDMSource.BDM;
import BDMSource.BDMData;
import BSSource.BlockSplit;
import BSSource.ReduceTask;

public class Main {
	
	private static int numReducesTask = 24;
	private static int partitions = 1;
	
	private static void runJobBDM(String filePath, String DirPath) throws Exception {
		
		Configuration conf = new Configuration();
	    Job job = new Job(conf, "BDM Execution");
	    
		job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(IntWritable.class);
	    
	    job.setMapperClass(BDM.Map.class);
	    job.setReducerClass(BDM.Reduce.class);
	    
	    job.setJarByClass(BDM.class);
	    job.setNumReduceTasks(3);
	        
	    job.setInputFormatClass(TextInputFormat.class);
	    job.setOutputFormatClass(TextOutputFormat.class);
	    
	    String[] patitions = new File(filePath).list();
	    for (String partition : patitions) {
	    	if (partition.substring(partition.length()-3, partition.length()).equalsIgnoreCase("txt")){
	    		FileInputFormat.addInputPath(job, new Path(filePath + partition));
	    	}
		}
	    FileOutputFormat.setOutputPath(job, new Path(DirPath));
	    
	    job.waitForCompletion(true);
	}
	
	private static void runJobSBP(String filePath, String DirPath) throws Exception {
		
		Configuration conf = new Configuration();
		conf.set("DirPath", filePath);
		conf.set("numReduceTasks", String.valueOf(numReducesTask));
		
		Job job = new Job(conf, "SlicedBlockPacker Execution");
	    
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(Text.class);
	    job.setNumReduceTasks(numReducesTask);
	    
	    job.setMapperClass(SlicedBlockPacker.Map.class);
	    job.setPartitionerClass(SlicedBlockPacker.Part.class);
	    job.setReducerClass(SlicedBlockPacker.Reduce.class);
	    job.setJarByClass(SlicedBlockPacker.class);
	    
	    for (int i = 0; i < partitions; i++) {
	    	FileInputFormat.addInputPath(job, new Path(filePath + "Partition_" + i + ".txt"));
		}
	    FileOutputFormat.setOutputPath(job, new Path(DirPath));
	    job.waitForCompletion(true);

	}
	
	private static void runJobSB(String filePath, String DirPath) throws Exception {
		
		Configuration conf = new Configuration();
		conf.set("DirPath", filePath);
		conf.set("numReduceTasks", String.valueOf(numReducesTask));
		conf.set("numMapTasks", String.valueOf(partitions));
		
		Job job = new Job(conf, "SlicedBlock Execution");
	    
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(Text.class);
	    job.setNumReduceTasks(numReducesTask);
	    
	    job.setMapperClass(SlicedBlock.Map.class);
	    job.setPartitionerClass(SlicedBlock.Part.class);
	    //job.setGroupingComparatorClass(BlockSplit.NaturalKeyGroupingComparator.class);
	    job.setReducerClass(SlicedBlock.Reduce.class);
	    job.setJarByClass(SlicedBlock.class);
	    
	    for (int i = 0; i < partitions; i++) {
	    	FileInputFormat.addInputPath(job, new Path(filePath + "Partition_" + i + ".txt"));
		}
	    FileOutputFormat.setOutputPath(job, new Path(DirPath));
	        
	    job.waitForCompletion(true);

	}
	
	private static void runJobBP(String filePath, String DirPath) throws Exception {
		
		Configuration conf = new Configuration();
		conf.set("DirPath", filePath);
		conf.set("numReduceTasks", String.valueOf(numReducesTask));
		conf.set("numMapTasks", String.valueOf(partitions));
		
		Job job = new Job(conf, "Block Split Execution");
	    
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(Text.class);
	    job.setNumReduceTasks(numReducesTask);
	    
	    job.setMapperClass(BlockSplit.Map.class);
	    job.setPartitionerClass(BlockSplit.Part.class);
	    //job.setGroupingComparatorClass(BlockSplit.NaturalKeyGroupingComparator.class);
	    job.setReducerClass(BlockSplit.Reduce.class);
	    job.setJarByClass(BlockSplit.class);
	    
	    for (int i = 0; i < partitions; i++) {
	    	FileInputFormat.addInputPath(job, new Path(filePath + "Partition_" + i + ".txt"));
		}
	    FileOutputFormat.setOutputPath(job, new Path(DirPath));
	        
	    job.waitForCompletion(true);

	}

	public static void main(String[] args) throws Exception {
		 Long initTime = System.currentTimeMillis();
		 //String path = "/user/cpd-dev-34/cpd/input/";
		 String path = "/user/dp-pc/demetrio/input/";
		 //String path = "D:/DataSets/DataSet2Splitted/"+partitions+"MapInputs/";
		 //String path = "D:/DataSets/Teste/";
		 
		 
		 String method = args[0];
		 partitions = Integer.parseInt(args[1]);
		 numReducesTask = Integer.parseInt(args[2]);
		 path = path + args[3];
		 
		 //String method = "sb";
		 
		 ////cygdrive/c/cygwin/usr/Tester/ZaidaExperimento
		 
		 if (method.equalsIgnoreCase("BDM")){
			 runJobBDM(path, path + "ZaidaExperimento");
		 }
		 
		 if (method.equalsIgnoreCase("SBP")){
			 configureSPBTasksAllocator(path);
			 runJobSBP(path, path + "ZaidaSlicedBlockPacker");	
		 }
		 
		 if (method.equalsIgnoreCase("SB")){
			 configureSBTasksAllocator(path);
			 runJobSB(path, path + "ZaidaSlicedBlock");	
		 }
		 
		 if (method.equalsIgnoreCase("BS")){
			 configureBSTasksAllocator(path);
			 runJobBP(path, path + "ZaidaBlockSplit");
		 }
		 
		 double d = (System.currentTimeMillis() - initTime);
		 System.out.println("Total Running: " + d/1000 + "s");
	 }
	
	private static void configureBSTasksAllocator(String dirPath) throws IOException, InterruptedException {
		getBDM(dirPath);
		BlockSplit.map_configure(new IntWritable(partitions), new IntWritable(numReducesTask));
		saveResultsOfBSAllocatorInDFS(dirPath);
	}
	
	private static void configureSBTasksAllocator(String dirPath) throws IOException, InterruptedException {
		getBDM(dirPath);
		SlicedBlock.map_configure(new IntWritable(partitions), new IntWritable(numReducesTask));
		saveResultsOfSBAllocatorInDFS(dirPath);
	}

	private static void configureSPBTasksAllocator(String dirPath) throws IOException, InterruptedException {
		getBDM(dirPath);
		SlicedBlockPacker.map_configure(new IntWritable(partitions), new IntWritable(numReducesTask));
		saveResultsOfSBPAllocatorInDFS(dirPath);
	}
	
	private static void getBDM(String dirPathBDM) {
		BDM.blockingPos = 0;
		BDM.sumPairs = 0;
		BDM.BDMData = new ArrayList<BDMData>();
		BDM.blockingkeyIndexer = new HashMap<String, Integer>();
    	try{
    		FileSystem fs = FileSystem.get(new Configuration());
			BufferedReader buffReader = new BufferedReader(new InputStreamReader(fs.open(new Path(dirPathBDM + "BDM.txt"))));
			HashMap<Integer, Integer> partitions = new HashMap<Integer, Integer>();
			int blockSize = 0;
			String line = buffReader.readLine();
			while (line != null){
				String blockingKey = line.substring(0, 3);
				int lastDelimiterIndex = line.lastIndexOf("\t");
				int blockPartSize = Integer.parseInt(line.substring(lastDelimiterIndex+1, line.length()));
				line = line.substring(0, lastDelimiterIndex);
				lastDelimiterIndex = line.lastIndexOf(".");
				int partition = Integer.parseInt(line.substring(lastDelimiterIndex+1, line.length()));
				partitions.put(partition, blockPartSize);
				blockSize += blockPartSize;
				line = buffReader.readLine();
				if (line == null || !blockingKey.equals(line.substring(0, 3))){
					if(blockSize > SlicedBlockPacker.largestBlockSize){
						SlicedBlockPacker.largestBlockSize = blockSize;
						SlicedBlock.largestBlockSize = blockSize;
					}
					BDM.BDMData.add(new BDMData(blockingKey, partitions, blockSize));
					BDM.blockingkeyIndexer.put(blockingKey, BDM.blockingPos);
					BDM.blockingPos++;
					BDM.sumPairs += (0.5) * blockSize * (blockSize-1);
					partitions = new HashMap<Integer, Integer>();
					blockSize = 0;
				}
			}
    	}catch(Exception e){
    		e.printStackTrace();
    	}
	}
	
	private static void saveResultsOfSBPAllocatorInDFS(String dirPath) {
		try{
    		FileSystem fs = FileSystem.get(new Configuration());
    		FSDataOutputStream writer = fs.create(new Path(dirPath + "matchTasks/AllocatedMatchTasks.txt"));
    		HashMap<String, AllocatedBlockData> allocatedMatchTasks = SlicedBlockPacker.allocatedMatchTasks;
    		Object[] keys = allocatedMatchTasks.keySet().toArray();
    		for (Object tmpKey : keys) {
    			writer.writeBytes(tmpKey + "\t" + allocatedMatchTasks.get(tmpKey).getBlockIndex() + "." +
    					allocatedMatchTasks.get(tmpKey).getReduceTaskIndex() + "." +
    					allocatedMatchTasks.get(tmpKey).getType() + "." +
    					allocatedMatchTasks.get(tmpKey).getNumberOfEntities() + "\n");
    		}
    		writer.close();
    		FSDataOutputStream writer2 = fs.create(new Path(dirPath + "matchTasks/AllocatedSlicedMatchTasks.txt"));
    		HashMap<String, ArrayList<AllocatedBlockData>> allocatedSlicedMatchTasks = SlicedBlockPacker.allocatedSlicedMatchTasks;
    		keys = allocatedSlicedMatchTasks.keySet().toArray();
    		for (Object tmpKey : keys) {
    			ArrayList<AllocatedBlockData> list = allocatedSlicedMatchTasks.get(tmpKey);
    			String strTmp = "";
    			for (AllocatedBlockData allocatedBlockData : list) {
    				strTmp = strTmp + "\t" + allocatedBlockData.getBlockIndex() + "." +
    					allocatedBlockData.getReduceTaskIndex() + "." +
    					allocatedBlockData.getType() + "." +
    					allocatedBlockData.getNumberOfEntities();
				}
    			writer2.writeBytes(tmpKey + "\t" + strTmp + "\n");
    		}
    		writer2.close();
    	}catch(Exception e){
    		e.printStackTrace();
    	}
	}
	
	private static void saveResultsOfSBAllocatorInDFS(String dirPath) {
		try{
    		FileSystem fs = FileSystem.get(new Configuration());
    		FSDataOutputStream writer = fs.create(new Path(dirPath + "matchTasks3/AllocatedMatchTasks.txt"));
    		HashMap<String, AllocatedBlockData> allocatedMatchTasks = SlicedBlock.allocatedMatchTasks;
    		Object[] keys = allocatedMatchTasks.keySet().toArray();
    		for (Object tmpKey : keys) {
    			writer.writeBytes(tmpKey + "\t" + allocatedMatchTasks.get(tmpKey).getBlockIndex() + "." +
    					allocatedMatchTasks.get(tmpKey).getReduceTaskIndex() + "." +
    					allocatedMatchTasks.get(tmpKey).getType() + "." +
    					allocatedMatchTasks.get(tmpKey).getNumberOfEntities() + "\n");
    		}
    		writer.close();
    		FSDataOutputStream writer2 = fs.create(new Path(dirPath + "matchTasks3/AllocatedSlicedMatchTasks.txt"));
    		HashMap<String, ArrayList<AllocatedBlockData>> allocatedSlicedMatchTasks = SlicedBlock.allocatedSlicedMatchTasks;
    		keys = allocatedSlicedMatchTasks.keySet().toArray();
    		for (Object tmpKey : keys) {
    			ArrayList<AllocatedBlockData> list = allocatedSlicedMatchTasks.get(tmpKey);
    			String strTmp = "";
    			for (AllocatedBlockData allocatedBlockData : list) {
    				strTmp = strTmp + "\t" + allocatedBlockData.getBlockIndex() + "." +
    					allocatedBlockData.getReduceTaskIndex() + "." +
    					allocatedBlockData.getType() + "." +
    					allocatedBlockData.getNumberOfEntities();
				}
    			writer2.writeBytes(tmpKey + "\t" + strTmp + "\n");
    		}
    		writer2.close();
    	}catch(Exception e){
    		e.printStackTrace();
    	}
	}
	
	private static void saveResultsOfBSAllocatorInDFS(String dirPath) {
		try{
    		FileSystem fs = FileSystem.get(new Configuration());
    		FSDataOutputStream writer = fs.create(new Path(dirPath + "matchTasksBS/MatchRedTasks.txt"));
    		Map<String, ReduceTask> matchRedTasks = BlockSplit.matchRedTasks;
    		Object[] keys = matchRedTasks.keySet().toArray();
    		for (Object tmpKey : keys) {
    			writer.writeBytes(tmpKey + "\t" + matchRedTasks.get(tmpKey).getIdReduceTask() + "." +
    					matchRedTasks.get(tmpKey).getComps() + "\n");
    		}
    		writer.close();
    	}catch(Exception e){
    		e.printStackTrace();
    	}
	}
}
